<?php
namespace Core\Common\Handler;

use Core\Common\Base\Config;
use Core\Common\Base\Log;

/**
 * 进程名与app.ini中的process.<进程名>相同, 都是小写
 * Class KafkamHandler
 * @package Core\Common\Handler
 */
class KafkamHandler
{
    const WAIT_QUEUE_EMPTY = 1;  // 停止时先等待队列排空
    const WAIT_CHILD_EXIT = 2;   // 停止时等待子进程退出

    private static $name = null;
    private static $timeout = null;
    private static $process = null;

    private static $executors = [];
    private static $stoping = 0;

    /**
     * 初始化属性及子进程
     */
    public static function init($process)
    {
        $tmp = explode('\\', __CLASS__);
        $processName = str_replace('Handler', '', array_pop($tmp));
        $parameter = Config::config('process.'.strtolower($processName).'.parameter');
        $arr = explode(',', $parameter);

        list($name, $timeout, $cnt) = array_map('trim', $arr);

        self::$name = $name;
        self::$timeout = $timeout;
        self::$process = $process;
        $msgKey = crc32(__CLASS__);

        // 启动子进程
        foreach (range(1, $cnt) as $i) {
            $subProcess = self::newExecutorProcess($i);
            $subProcess->useQueue($msgKey, 2);//$process->pid
            if ($pid = $subProcess->start()) {
                self::$executors[$pid] = $subProcess;
            }
        }

        pcntl_signal(SIGUSR1, function ($signo) {
            self::$stoping = self::WAIT_QUEUE_EMPTY;
        });
        pcntl_signal(SIGTERM, function ($signo) {
            self::$stoping = self::WAIT_QUEUE_EMPTY;
        });
        pcntl_signal(SIGCHLD, function ($signo) {
            while($ret = \swoole_process::wait(false)) {
                unset(self::$executors[$ret['pid']]);
            }
        });
    }

    public static function run(\swoole_server $server, $process)
    {
        if (is_null(self::$name)) {
            if (!class_exists('App\Consumer\CommonConsumer')) {
                Log::error(LOG_INFO, ['message' => "App\Consumer\CommonConsumer not found", ]);
                return 'no CommonConsumer';
            }
            self::init($process);
        }

        $kafka = \Core\Common\Base\Kafka::getInstance(self::$name);
        try {
            // 等待队列排空
            if (self::$stoping == self::WAIT_QUEUE_EMPTY) {
                $queueNum = self::getQueueNum();
                Log::process(LOG_DEBUG, [
                    'message' => 'WAIT_QUEUE_EMPTY', 'pid' => self::$process->pid, 'queue_num' => $queueNum
                ]);
                if ($queueNum == 0) {
                    self::$stoping = self::WAIT_CHILD_EXIT;
                    foreach (self::$executors as $executor) {
                        #停止时禁止清空 queue msg
//                        $executor->freeQueue();
                        posix_kill($executor->pid, SIGTERM);
                    }
                }
            }

            // 等待子进程退出
            if (self::$stoping == self::WAIT_CHILD_EXIT) {
                $executorCount = count(self::$executors);
                Log::process(LOG_DEBUG, [
                    'message' => 'WAIT_CHILD_EXIT', 'pid' => self::$process->pid, 'executor_count' => $executorCount
                ]);
                if (0 === $executorCount) {
                    self::$process->exit(1);
                }
            }

            // 队列中有待处理的任务，就退出等待下一次轮训
            if (self::$stoping == 0 ) {//&& self::getQueueNum() < count(self::$executors)
                $message = $kafka->consume(self::$timeout);
                if (!$message) {
                    return '';
                }

                $first = current(self::$executors);
                $first->push(
                    igbinary_serialize(['key' => $message->key, 'value' => $message])
                );
                return true;
            }

        } catch (\Throwable $ex) {
            $detail = [
                'error_type' => 'KafkaHandler',
                'error_source' => self::$name,
                'query' => [],
                'error_code' => $ex->getCode(),
                'message' => $ex->getMessage(),
                'trace' => $ex->getTrace(),
            ];
            Log::error(LOG_ERR, $detail);
            return 0;
        }
        return '';
    }

    /**
     * 获取当前队列任务大小
     * @return int
     */
    public static function getQueueNum()
    {
        $first = current(self::$executors);
        if ($first) {
            $stat = $first->statQueue();
            return $stat['queue_num'];
        }
        return 0;
    }

    /**
     * 创建子进程
     * @param $index
     * @return \swoole_process
     */
    public static function newExecutorProcess($index)
    {
        return new \swoole_process(function($process) use ($index) {
            if ('Linux' == PHP_OS) {//进程名
                $process->name(CORE_NAME . 'Service:kafkam Exec Process');
            }

            $stoping = false;

            pcntl_signal(SIGTERM, function ($signo) use ($process, $index, &$stoping) {//退出信号
                $stoping = true;
            });

            while (!$stoping) {
                pcntl_signal_dispatch();

                $msg = $process->pop();
                if ($msg === false) {
                    usleep(100);
                    continue;
                }
                usleep(100);
                $data = igbinary_unserialize($msg);
                call_user_func(['App\Consumer\CommonConsumer', 'run'], $data);
            }
        });
    }
}